package com.swak.mmap;

import java.io.File;
import java.io.IOException;
import java.text.NumberFormat;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.reactivex.threads.EventLoop;
import com.swak.reactivex.threads.SingleThreadEventLoop;
import com.swak.reactivex.threads.SwakThreadFactory;
import com.swak.reactivex.threads.waitstrategy.BlockingWaitStrategy;
import com.swak.utils.Maps;
import com.swak.utils.ring.RingBuffer;

import lombok.Getter;
import lombok.Setter;
import lombok.experimental.Accessors;

/**
 * MappedFileAllocator.
 * 
 * @author 618lf
 */
public class MappedFileAllocator {

	private static final SwakThreadFactory Factory = new SwakThreadFactory("MappedFile.Allocator-", false,
			new AtomicInteger());
	private static final int PreAllocateSize = 2;

	private static final Logger log = LoggerFactory.getLogger(MappedFileAllocator.class);
	private static final long WAIT_TIME = 1000 * 5;
	private final EventLoop eventLoop;
	private boolean customEventLoop = true;
	private final Map<String, AllocateTask> allocates = Maps.newHashMap();

	public MappedFileAllocator() {
		this(new SingleThreadEventLoop(Factory, new BlockingWaitStrategy()));
		this.customEventLoop = false;
	}

	public MappedFileAllocator(EventLoop eventLoop) {
		this.eventLoop = eventLoop;
	}

	public synchronized void registry(AllocateTask config) {
		if (!this.eventLoop.inEventLoop()) {
			this.eventLoop.execute(() -> {
				allocates.put(config.getStorePath(), config);
			});
		} else {
			allocates.put(config.getStorePath(), config);
		}
	}

	public MappedFile allocate(String storePath) {
		CompletableFuture<MappedFile> future = new CompletableFuture<MappedFile>();
		this.eventLoop.execute(() -> {
			this.doAllocate(storePath, future);
		});
		try {
			return future.get(WAIT_TIME, TimeUnit.MILLISECONDS);
		} catch (Exception e) {
			log.error("wait for allocate mapped file time out:", e);
		}
		return null;
	}

	private void doAllocate(String storePath, CompletableFuture<MappedFile> future) {

		/*
		 * 分配配置
		 */
		AllocateTask config = allocates.get(storePath);
		if (config == null) {
			future.complete(null);
			return;
		}

		/*
		 * 分配的缓存
		 */
		MappedFile file = config.getRing().pop();
		try {

			// 分配一个
			if (file == null) {
				file = this.doAllocate(config);
			}

			// 预分配一个
			this.preAllocate(config);

		} finally {
			// 返回数据
			future.complete(file);
		}
	}

	private MappedFile doAllocate(AllocateTask config) {
		MappedFile file = null;
		try {
			String path = config.storePath + File.separator
					+ config.mappedFileNaming.format(this.offset2FileName(config.offset));
			file = new MappedFile(path, config.offset, config.fileSize, config.maxRecordSize, config.flushLeastPages);
			file.recover().warmMappedFile(FlushDiskType.ASYNC_FLUSH, 1024 / 4 * 16);
			config.offset++;
		} catch (Exception e) {
			log.error("allocate mapped file error.", e);
			if (file != null) {
				try {
					file.destroy(1000);
				} catch (IOException e1) {
				}
			}
		}
		return file;
	}

	private String offset2FileName(final long offset) {
		final NumberFormat nf = NumberFormat.getInstance();
		nf.setMinimumIntegerDigits(20);
		nf.setMaximumFractionDigits(0);
		nf.setGroupingUsed(false);
		return nf.format(offset);
	}

	private void preAllocate(AllocateTask config) {
		MappedFile file = this.doAllocate(config);
		if (file != null) {
			config.ring.add(file);
		}
	}

	public void shutdown() {
		try {
			if (this.eventLoop != null && this.customEventLoop) {
				this.eventLoop.shutdownGracefully().get();
			}
		} catch (Exception e) {
		}
		try {
			this.allocates.forEach((k, e) -> {
				try {
					e.ring.shutdown(1000);
				} catch (Exception e2) {
				}
			});
		} catch (Exception e) {
		}
	}

	/**
	 * 分配配置
	 * 
	 * @author 618lf
	 */
	@Getter
	@Setter
	@Accessors(chain = true)
	public static class AllocateTask {
		private String storePath;
		private int fileSize;
		private int maxRecordSize;
		private int flushLeastPages;
		private MappedFileNaming mappedFileNaming;
		private final Ring ring;
		private int offset;

		public AllocateTask() {
			this.ring = new Ring(PreAllocateSize);
		}

		public AllocateTask(int preAllocateSize) {
			this.ring = new Ring(preAllocateSize);
		}
	}

	/**
	 * 这两个数不能相等.
	 * 写的时候不能走到 read 的位置.
	 * 
	 * @author 618lf
	 */
	static class Ring extends RingBuffer<MappedFile> {

		long read = 0;
		long write = 0;

		Ring(int allocateSize) {
			super(allocateSize);
		}

		boolean isFull() {
			return this.isFull(read, write);
		}

		boolean isEmpty() {
			return this.isEmpty(read, write);
		}

		void add(MappedFile request) {
			if (this.isFull()) {
				throw new IndexOutOfBoundsException();
			}
			this.put(write++, request);
		}

		MappedFile pop() {
			if (this.isEmpty()) {
				return null;
			}
			return this.get(read++);
		}

		void shutdown(int intervalForcibly) {
			for (;;) {
				try {
					MappedFile file = pop();
					if (file != null) {
						file.shutdown(intervalForcibly);
					}
				} catch (Exception e) {
				}

			}
		}
	}
}